package com.batchengine.worker;

import com.batchengine.BatchEngineException;
import com.batchengine.Definitions;
import com.batchengine.task.Task;
import com.batchengine.task.TaskContext;
import static com.batchengine.task.TaskUtil.getTaskContext;
import static com.batchengine.util.LoggingFactory.createLogger;
import static com.batchengine.worker.Status.FINISHED;
import static com.batchengine.worker.Status.RUNNING;
import org.apache.log4j.Logger;

import static java.text.MessageFormat.format;
import java.util.concurrent.BlockingQueue;

/**
 * User: sunil.incubator@yahoo.com
 * Date: Sep 16, 2009
 * Time: 2:45:11 PM
 */
public class ScheduledDataConsumer extends DataConsumer {

    private static final Logger logger = createLogger();

    public ScheduledDataConsumer(String name, long taskId, BlockingQueue workQ) {
        super(name, taskId, workQ);
    }

    public ScheduledDataConsumer(String name, long taskId, ThreadGroup workGroup, BlockingQueue workQ) {
        super(name, taskId, workGroup, workQ);
    }

    public ScheduledDataConsumer(String name, long taskId, ThreadGroup workGroup, int workerPriority, BlockingQueue workQ) {
        super(name, taskId, workGroup, workerPriority, workQ);
    }

    @Override
    public void run() {

        try {

            workerThread = Thread.currentThread();
            try {

                setStatus(RUNNING);
                doWork(getTaskContext(getJobId()));
            } catch (Exception ignored) {
                logger.error(format("error running the ScheduledDataConsumer {0}", getName()), ignored);
            }
        } finally {
            cleanUp();
            setStatus(FINISHED);
        }
    }

    public void doWork(TaskContext context) throws InterruptedException {

        if (null == getDataGatherer()) {
            logger.error("No gatherer configured " + this);
        }

        try {

            while (!FINISHED.equals(getDataGatherer().getStatus())) {
//                logger.debug("------------------------gatherer not yet finished---------------------");
                Thread.sleep(2000);
            }
            while (0 < workQ.size()) {
                Task task = (Task) context.getContextParam(Definitions.ContextParams.TASK_INSTANCE);
                context.addcontextParam(Definitions.ContextParams.ARGUMENTS, workQ.poll());
                context.addcontextParam(Definitions.ContextParams.OWNER, this);
                task.execute(context);
            }
        } catch (Exception e) {
            throw new BatchEngineException("error#ScheduledDataConsumer@doWork", e);
        }
    }
}